package ssh

import (
	"context"
	"fmt"
	"github.com/sirupsen/logrus"
	"github.com/tluo-github/super-runner/common"
	"github.com/tluo-github/super-runner/executors"
	"github.com/tluo-github/super-runner/helpers/kafka"
	"github.com/tluo-github/super-runner/helpers/ssh"
	"github.com/tluo-github/super-runner/helpers/minio"
	"os"
	"path/filepath"
	"time"
)
type SshExecutor struct {
	Build       common.Build
	Config common.RunnerConfig
	buildFinish chan error
	BuildLog    *os.File
	Context context.Context
	sshName string
	sshCommand [] *ssh.Client
	ObjectName string
	IsSystemError bool

}


func (s *SshExecutor) Prepare(config common.RunnerConfig, build common.Build) error {
	s.Config = config
	s.Build = build
	s.buildFinish = make(chan error, 1)
	s.IsSystemError = false

	s.sshName = fmt.Sprintf("super-runner-%s-%d",s.Build.Name,s.Build.Timestamp)
	// 创建 build log
	filename := fmt.Sprintf("/logs/%s.log",s.sshName)
	build_log, err := os.Create(filename)
	if err != nil {
		return err
	}
	s.BuildLog = build_log
	timeStr:=time.Now().Format("2006-01-02")
	s.ObjectName = fmt.Sprintf("%s/%s",timeStr,filepath.Base(s.BuildLog.Name()))


	logrus.WithFields(logrus.Fields{
		"name": s.sshName,
	}).Infoln("Prepare")

	// 发送开始执行消息给 kafka
	kafka.StartKafka(config.KafkaConfig,s.sshName,build.MetaData,config.Name)

	// Create SSHs command
	for _,clientConfig := range s.Build.SshConfig{
		s.sshCommand = append(s.sshCommand,&ssh.Client{
			SshConfig:      clientConfig,
			Stdout:         s.BuildLog,
			Stderr:         s.BuildLog,
		})
	}

	s.Context = context.Background()

	logrus.Infoln("Connecting to SSH server...")
	for _,client := range s.sshCommand{
		err = client.Connect()
		logrus.Infof("ip : %s, port: %s, user : %s, password : %s, idfile: %s",client.Host,client.Port,client.User,client.Password,client.IdentityFile)
		if err != nil {
			s.IsSystemError = true
			return fmt.Errorf("ssh command Connect() error: %w", err)
		}
	}
	return nil
}

func (s *SshExecutor) Run() error {
	logrus.WithFields(logrus.Fields{
		"name": s.sshName,
	}).Infoln("Run")

	var err error
	for _,client := range s.sshCommand {
		err = client.Run(s.Context,ssh.Command{
			Environment: nil,
			Command:     []string{"/bin/sh"},
			Stdin:       s.Build.Command,
		})
		if err != nil{
			break
		}
	}
	return err
}

func (s *SshExecutor) Wait() error {
	url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
	kafka.SucceededKafka(s.Config.KafkaConfig,s.sshName,s.Build.MetaData,url)
	return nil
}

func (s *SshExecutor)SendError(err error)  {
	logrus.WithFields(logrus.Fields{
		"name": s.sshName,
		"error": err,
	}).Infoln("SendError")

	url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
	kafka.ErrorKafka(s.Config.KafkaConfig,s.sshName,s.Build.MetaData,url,err,s.IsSystemError)
	s.BuildLog.WriteString(err.Error())
}


func (s *SshExecutor) Cleanup() error {

	logrus.WithFields(logrus.Fields{
		"name": s.sshName,
	}).Infoln("Cleanup")

	for _,client := range s.sshCommand {
		client.Cleanup()
	}
	s.cleanBuildLog()

	return nil

}

func (s *SshExecutor) cleanBuildLog()  {
	if s.BuildLog != nil {
		if minio.NewMinioHelper(s.Config.MinioConfig) != nil{
			_,err := minio.NewMinioHelper(s.Config.MinioConfig).PutObject(s.ObjectName,s.BuildLog.Name())
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"name": s.sshName,
				}).Errorf("Minio PutObject with error: %v", err)
			}
			logrus.WithFields(logrus.Fields{
				"name": s.sshName,
				"objectName": s.ObjectName,
			}).Infoln("Minio PutObject ok")
		}
		os.Remove(s.BuildLog.Name())
	}
}

func createFn() common.Executor {
	return &SshExecutor{}
}
func init()  {
	common.RegisterExecutor("ssh",executors.DefaultExecutorProvider{Creator: createFn})
}
